package streaming.api.transform;

import streaming.api.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 分流
 * Split 和 Select (不使用)
 * process (现在使用)
 *
 */
public class TransformTest4 {

    public static void main(String[] args) throws Exception {
        /**运行环境*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> inputStream = env.readTextFile("D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt");
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]), new Double(fields[2]));
        });
        /**1、定义OutputTag*/
        OutputTag<SensorReading> sideOutputTag = new OutputTag<SensorReading>("SensorReadingOutputTag"){};

        /**2、在ProcessFunction中处理主流和分流*/
        SingleOutputStreamOperator<SensorReading> processedStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {

            @Override
            public void processElement(SensorReading sensorReading, Context context, Collector<SensorReading> collector) throws Exception {
                //侧流-只输出特定数据
                if (sensorReading.getTemperature() > 30) {
                    context.output(sideOutputTag, sensorReading);
                }
                //主流
                else {
                    collector.collect(sensorReading);
                }
            }
        });
        //获取主流
        processedStream.print("main:");
        //获取侧流
        processedStream.getSideOutput(sideOutputTag).print("side:");
        env.execute();
    }
}
